package com.xiaotu.code.system;

import java.util.ArrayList;
import java.util.List;

import javax.jms.JMSException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.listener.DefaultMessageListenerContainer;

import com.xiaotu.code.model.quene.SettingMqQuene;
import com.xiaotu.code.service.mquene.SettingMqQueneService;

public class QueneSubscriberContainer implements InitializingBean, DisposableBean {

	protected final Logger logger = LoggerFactory.getLogger(getClass());
	private boolean enabled = true;

	@Autowired
	private SettingMqQueneService settingMqQueneService;
	private List<DefaultMessageListenerContainer> containerList = new ArrayList<DefaultMessageListenerContainer>();
	// 消费者列表
	private List<QueneMessageConsumer> consumers = null;
	// 连接工厂容器
	private JmsConnectionFactoryContainer jmsConnectionFactoryContainer;

	/**
	 * 订阅节点队列.
	 * 
	 * @param nodes
	 * @throws ServiceException
	 * @throws JMSException
	 */
	private void subscribeNodeTopic(SettingMqQuene quene) throws ServiceException, JMSException {

		// 侦听的节点列表
		String strSendNodes = quene.getSendNodes();
		String[] arrayNodes = strSendNodes.split(",");
		int currentNodeId = 0;
		boolean enabledLocal = quene.getEnabledLocal();
		String queneName = quene.getName();
		for (String strNode : arrayNodes) {
			// 非本节点
			int nodeId = Integer.valueOf(strNode);
			if (!enabledLocal && currentNodeId == nodeId) {
				continue;
			}
		
			String url = "tcp://127.0.0.1:61616";
			QueneMessageConsumer consumer = getConsumer(queneName);
			if (consumer == null) {
				continue;
			}
			// queueName形如 nodeId-queneNmae-currentNodeId
			String reBuildQueneName = QueneSubscriberContainer.generalQueneName(nodeId, queneName, currentNodeId);

			// 连接缓存工厂
			CachingConnectionFactory cacheFactory = this.jmsConnectionFactoryContainer.getConnectFactory(url);
			DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
			container.setConnectionFactory(cacheFactory);
			container.setMessageListener(consumer);
			container.setDestinationName(reBuildQueneName);
			// 并发消费者数，数据大于1可能会收到相同的消息.
			// container.setConcurrentConsumers(20);
			container.afterPropertiesSet();
			container.start();
			containerList.add(container);
			logger.info("queneSubscriberService启动队列监听成功: localNodeId={},queneName={},sendNodeId={},tcpUrl={}",
					new Object[] { currentNodeId, queneName, nodeId, url });
		}
	}

	/**
	 * 构造队列名称.
	 * 
	 * @param fromNode
	 * @param queneName
	 * @param toNode
	 * @return
	 */
	public static String generalQueneName(int fromNode, String queneName, int toNode) {

		return fromNode + "-" + queneName + "-" + toNode;
	}

	private QueneMessageConsumer getConsumer(String queneName) throws ServiceException {
		for (QueneMessageConsumer cons : consumers) {
			if (cons.getDestinationName().equalsIgnoreCase(queneName)) {
				return cons;
			}
		}
		// TODO:正式运行后，把抛出异常的放开lqh
		return null;
		// throw new ServiceException("找不到队列的" + queneName + "的消费者.");
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		if (this.enabled) {
			// 初始化队列监听.
			try {

				List<SettingMqQuene> quenes = this.settingMqQueneService.getLitenerList();
				for (SettingMqQuene quene : quenes) {
					if (quene.getEnabled()) {
						this.subscribeNodeTopic(quene);
					}
				}
				logger.info("#####化队列监听已初始化成功.");
			} catch (Exception e) {
				throw e;
			}
		}
	}

	@Override
	public void destroy() throws Exception {
		for (DefaultMessageListenerContainer container : containerList) {
			container.destroy();
		}

	}

	public boolean isEnabled() {
		return enabled;
	}

	public void setEnabled(boolean enabled) {
		this.enabled = enabled;
	}

	public List<QueneMessageConsumer> getConsumers() {
		return consumers;
	}

	public void setConsumers(List<QueneMessageConsumer> consumers) {
		this.consumers = consumers;
	}

	public void setJmsConnectionFactoryContainer(JmsConnectionFactoryContainer jmsConnectionFactoryContainer) {
		this.jmsConnectionFactoryContainer = jmsConnectionFactoryContainer;
	}

}
